#include "Xps/Protocol/UDP/Multicast.h"

namespace Xps
{
    namespace UDP
    {
        Multicast::Multicast( const std::string &_Name, const ip::address &_ListenAddress, const ip::address &_MulticastAddress, const short &_Port, const IObjectFactory *_ObjectFactory, const bool &_IsLoopBackEnabled /*= false*/)
            :name(_Name),multicast_endpoint(_MulticastAddress,_Port), listen_endpoint(_ListenAddress,_Port), is_loopback(_IsLoopBackEnabled), object_factory(_ObjectFactory), is_running(false)
        {
            Initialize();
        }

        const std::string & Multicast::getName() const
        {
            return name;
        }

        void Multicast::Start()
        {
            if(!is_running)
            {
                is_running = true;
                io_service = boost::shared_ptr<asio::io_service>(new asio::io_service());
                socket = boost::shared_ptr<udp::socket>(new udp::socket(*io_service));
                socket->open(listen_endpoint.protocol());
                socket->set_option(udp::socket::reuse_address(true));
                socket->set_option(ip::multicast::enable_loopback(is_loopback));
                socket->bind(listen_endpoint);
                socket->set_option(ip::multicast::join_group(multicast_endpoint.address()));

                m_reciever = boost::thread(&Multicast::Recieve,this);
                th_sender = boost::thread(&Multicast::Sender,this);
                th_process_packets = boost::thread(&Multicast::ProcessPackets, this);
            }
        }

        void Multicast::Stop()
        {
            if(is_running)
            {
                is_running = false;
                io_service->stop();
                TerminateThread(m_reciever.native_handle(),0);
                TerminateThread(th_sender.native_handle(),0);
                TerminateThread(th_process_packets.native_handle(),0);
                socket->shutdown(tcp::socket::shutdown_both);
                socket->close();
            }
        }

        void Multicast::Send( IPacket::Ptr _Packet )
        {
            outgoing_queue.push(_Packet);
            SetEvent(sender_event);
        }

        void Multicast::Initialize()
        {
            recieve_event = CreateEvent(NULL, FALSE, FALSE, NULL);
            sender_event = CreateEvent(NULL, FALSE, FALSE, NULL);
        }

        void Multicast::Recieve()
        {
            static byte byte_buffer[MAX_PACKET_SIZE];
            udp::endpoint local_endpoint = socket->local_endpoint();
            udp::endpoint sender_endpoint;
            while(is_running)
            {
                memset(byte_buffer,0,MAX_PACKET_SIZE);
                try
                {
                    long length = socket->receive_from(asio::buffer(byte_buffer,MAX_PACKET_SIZE),sender_endpoint);
                    IObject::Ptr object = object_factory->FromBytes(byte_buffer,length);
                    incomming_queue.push(object);
                    SetEvent(recieve_event);
                }
                catch(...)
                {
                    Stop();
                }

            }
        }

        void Multicast::Sender()
        {
            static byte byte_buffer[MAX_PACKET_SIZE];
            while(is_running)
            {
                WaitForSingleObject(sender_event,INFINITE);
                for(int i=0,
                    size = outgoing_queue.size();
                    i<size;i++)
                {
                    IPacket::Ptr packet = outgoing_queue.pop();
                    
                    memset(byte_buffer,0,MAX_PACKET_SIZE);
                    try
                    {
                        long length = object_factory->ToBytes(packet->getPayload(), byte_buffer, MAX_PACKET_SIZE);
                        socket->send_to(asio::buffer(byte_buffer,length),multicast_endpoint);
                    }
                    catch(...)
                    {
                        Stop();
                    }

                }
            }
        }

        void Multicast::ProcessPackets()
        {
            while(is_running)
            {
                WaitForSingleObject(recieve_event, INFINITE);
                for(int i=0,
                    size = incomming_queue.size();
                    i<size;i++)
                {
                    const IObject::Ptr object = incomming_queue.pop();
                    __raise OnPacketRecieved(name, object.get());

                }
            }
        }

        Multicast::~Multicast( void )
        {
            Stop();
            CloseHandle(recieve_event);
            CloseHandle(sender_event);
        }
    };
};